package it.learn.kafka.init;

import org.reactivestreams.Publisher;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.cloud.gateway.route.Route;
import org.springframework.cloud.gateway.route.RouteLocator;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

// @Component
public class Test01 implements CommandLineRunner {

    @Autowired
    private RouteLocator routeLocator;

    @Override
    public void run(String... args) throws Exception {

        Flux<Route> routes = routeLocator.getRoutes();

        Mono<Route> concatMap = routes.concatMap(route -> {
                    Mono<Route> routeMono = Mono.just(route)
                            .filterWhen(r -> {
                                System.out.println("r.getId() = " + r.getId());
                                Publisher<Boolean> apply = r.getPredicate().apply(null);
                                return apply;
                            })
                            .doOnError(e -> {
                                System.out.println("Error applying predicate for route: " + route.getId());
                                e.printStackTrace();
                            })
                            .onErrorResume(e -> Mono.empty());
                    return routeMono;
                })
                .next()
                .map(route -> {
                    System.out.println();
                    return route;
                });

        concatMap.subscribe(System.out::println);

        System.out.println("初始化成功......");
    }
}
